package com.pet.article.kafka.listener;

import com.pet.article.service.ApArticleService;
import com.pet.common.constants.MQConstants;
import com.pet.model.article.pojos.ArticleVisitStreamMsg;
import com.pet.model.behavior.dtos.UpdateArticleMsg;
import com.pet.utils.common.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Objects;

/**
 * 热点文章更新监听器
 */
@Component
@Slf4j
public class HotArticleHandlerListener {
    @Autowired
    private ApArticleService apArticleService;

    @KafkaListener(topics = MQConstants.HOT_ARTICLE_INPUT_TOPIC)
    public void handleHotArticle(ConsumerRecord<String, String> consumerRecord) {
        log.info("开始热点文章的更新 {}", consumerRecord.value());
        UpdateArticleMsg updateArticleMsg = JsonUtils.toBean(consumerRecord.value(), UpdateArticleMsg.class);
        ArticleVisitStreamMsg msg = new ArticleVisitStreamMsg();
        switch (Objects.requireNonNull(updateArticleMsg).getType().name()) {
            case "COMMENT":
                msg.setArticleId(updateArticleMsg.getArticleId());
                msg.setComment(1L);
                break;
            case "LIKES":
                msg.setArticleId(updateArticleMsg.getArticleId());
                msg.setLike(1L);
                break;
            case "COLLECTION":
                msg.setArticleId(updateArticleMsg.getArticleId());
                msg.setCollect(1L);
                break;
            case "VIEWS":
                msg.setArticleId(updateArticleMsg.getArticleId());
                msg.setView(1L);
                break;
        }

        apArticleService.updateArticle(msg);
    }
}
